package gupao.concurrency.demo.risk.scru;

import gupao.concurrency.demo.risk.exct.ExctResult;
import gupao.concurrency.demo.risk.model.ProcessDataModel;
import gupao.concurrency.demo.risk.scru.processor.ScruProcessor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Service
public class ScruService {

    @Autowired
    private ProcessDataModel processDataModel;

    private volatile boolean stop = false;

    private final AtomicInteger threadIdx = new AtomicInteger(0);

    private volatile Thread scruThread;

    @PostConstruct
    public void start(){
        createNewThread();
    }

    @PreDestroy
    public void stop(){
        stop = true;
        if(scruThread != null){
            scruThread.interrupt();
        }
    }

    private void createNewThread(){
        scruThread = new Thread(() -> {
            log.info("Scru service started.");
            while(!Thread.currentThread().isInterrupted() && !stop){
                try {
                    log.info("retrive exct result from quque..");
                    ExctResult lastResult = processDataModel.getExctResultsQueue().poll(10, TimeUnit.SECONDS);
                    log.info("retrive exct result from quque..{}", (lastResult == null ? null : lastResult.getJob().getId()));
                    if(lastResult != null){
                        ScruResult result = ScruProcessor.process(lastResult);
                        processDataModel.getScruResultsQueue().put(result);
                    }
                } catch (InterruptedException e) {
                    log.info("{} interrupted, create a new thread", Thread.currentThread().getName());
                    Thread.currentThread().interrupt();
                } catch (Throwable t){
                    log.error("error while do scru.", t);
                    //省略：记录audit table 便于后续重试
                    break;
                }
            }
            if(!stop){
                createNewThread();
            }
        }, "scru-thread-" + threadIdx.incrementAndGet());
        scruThread.start();
    }

}
